package com.atguigu.flink.window;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created by Smexy on 2023/2/27
 *
 *      滚动(增量)聚合： 窗口中每进入一个元素，就触发一次聚合。
 *                      允许输入和输出不是同一类型
 *
 */
public class Demo7_Aggregate
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //求每种传感器的水位和
                env
                   .socketTextStream("hadoop103", 8888)
                   .map(new WaterSensorMapFunction())
                   .keyBy(WaterSensor::getId)
                   .countWindow(3)
                   //等窗口关闭后，才会把数据发往下游的print
                   // <IN, ACC, OUT>
                   .aggregate(new AggregateFunction<WaterSensor, Integer, Integer>()
                   {
                       //创建累加器
                       @Override
                       public Integer createAccumulator() {
                           return 0;
                       }

                       //把来的一条数据进行累加
                       @Override
                       public Integer add(WaterSensor value, Integer accumulator) {
                           System.out.println("Demo7_Aggregate.add");
                           return accumulator += value.getVc();
                       }

                       //返回输出的结果
                       @Override
                       public Integer getResult(Integer accumulator) {
                           return accumulator;
                       }

                       //DataStreamAPI不需要写，DataSetAPI，才需要写
                       @Override
                       public Integer merge(Integer a, Integer b) {
                           return null;
                       }
                   })
                   .print();


                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }


    }
}
